

package caiqr.model.fb_asia_odds

//import com.db.five_million.football_match_sporttery_service
//import com.spark.firstApp.SavePrediction
//import com.hdfs.football_match_asia_odds
//import com.spark.firstApp.SavePrediction
import org.apache.spark.rdd.RDD
//import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SQLContext, DataFrame}
import java.text.SimpleDateFormat
import java.sql.DriverManager
//import java.lang.IllegalArgumentException


// 比较亚盘初盘/终盘盘口相同赛事
object AsiaInitOrCurrOdds {

  //1. 按照公司、赛季ID(2994,3444)分组统计, 初盘盘口相同的赛事, 按照时间倒叙排序, 汇总 match_id, match_result, asia_init_win_flag(初盘赢输盘)
  def asia_init_odds_same_by_company_seasonid(asia_df: DataFrame, mini_cnt: Int) = {
    // 1). 获取所有数据
    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "season_id", "season_name_pre", "season_pre", "init_odds", "match_time", "match_id","match_result", "asia_init_win_flag")

    val dds_type = "odds_init_sid"
    val new_match_id_list_rdd = compute_initodds_by_company_seasonid(match_list_rdd, dds_type, mini_cnt)

    new_match_id_list_rdd
  }


  //2. 按照公司、赛事(例如:英超,中超...)分组统计, 初盘盘口相同的赛事, 按照时间倒叙排序, 汇总 match_id, match_result, asia_init_win_flag(初盘赢输盘)
  def asia_init_odds_same_by_seasonname(asia_df: DataFrame, mini_cnt: Int) = {

    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "season_pre", "init_odds", "match_time", "match_id","match_result", "asia_init_win_flag")

    val dds_type = "odds_init_sname"
    val new_match_id_list_rdd = compute_initodds_by_company_seasonname(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }





  //3. 按照公司、赛季ID(2994,3444)分组统计, 终盘盘口相同的赛事, 按照时间倒叙排序, 汇总 match_id, match_result, asia_curr_win_flag(初盘赢输盘)
  def asia_curr_odds_same_by_company_seasonid(asia_df: DataFrame, mini_cnt: Int) = {
    // 1). 获取所有数据
    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "season_id", "season_name_pre", "season_pre", "curr_odds", "match_time", "match_id","match_result", "asia_curr_win_flag")

    val dds_type = "odds_curr_sid"
    val new_match_id_list_rdd = compute_initodds_by_company_seasonid(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }




  //4. 按照公司、赛事(例如:英超,中超...)分组统计, 终盘盘口相同的赛事, 按照时间倒叙排序, 汇总 match_id, match_result, asia_curr_win_flag(初盘赢输盘)
  def asia_curr_odds_same_by_seasonname(asia_df: DataFrame, mini_cnt: Int) = {

    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "season_pre", "curr_odds", "match_time", "match_id","match_result", "asia_curr_win_flag")

    val dds_type = "odds_curr_sname"
    val new_match_id_list_rdd = compute_initodds_by_company_seasonname(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }


  //5. 初盘和终盘盘口相同, 按照公司、赛季ID(2994,3444)分组统计, 汇总 match_id, match_result,asia_init_win_flag, asia_curr_win_flag(初终盘赢输盘)
  def asia_init_curr_odds_same_by_company_seasonid(asia_df: DataFrame, mini_cnt: Int) = {

    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "season_id", "season_name_pre", "season_pre", "init_odds", "curr_odds", "match_time", "match_id","match_result", "asia_init_win_flag", "asia_curr_win_flag")

    val dds_type = "odds_init_curr_sid"
    val new_match_id_list_rdd = compute_odds_by_company_seasonid(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }

  //6. 按照公司、赛事(例如:英超,中超...)分组统计, 初终盘盘口相同的赛事, 按照时间倒叙排序, 汇总 match_id, match_result,asia_init_win_flag, asia_curr_win_flag(初终盘赢输盘)
  def asia_init_curr_odds_same_by_seasonname(asia_df: DataFrame, mini_cnt: Int) = {

    val match_list_rdd = asia_df.orderBy(asia_df("match_time").asc).
      selectExpr("company_id", "season_pre", "init_odds", "curr_odds", "match_time", "match_id","match_result", "asia_init_win_flag", "asia_curr_win_flag")

    val dds_type = "odds_init_curr_sname"
    val new_match_id_list_rdd = compute_odds_by_seasonname(match_list_rdd, dds_type, mini_cnt)
    //new_match_id_list_rdd.collect().foreach(println)

    new_match_id_list_rdd
    // TODO
    // save to DB...
  }




  // 根据key(companyid_seasonid_seasonnamepre_seasonpre_initodds)
  // 统计后面参数(例如: match_id,match_result,asia_init_win_flag)
  def compute_initodds_by_company_seasonid(match_list_rdd: DataFrame, dds_type: String, mini_cnt: Int): RDD[(String,String)] ={

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    // 转换为元祖 (companyid_seasonid_seasonnamepre_seasonpre_initodds, (match_id,match_result,asia_init_win_flag,match_time) )
    // OUT: (280_687_2005/2006_世外欧洲_25,(130330,1,3,1126108740000))
    val tuple_same_init_odds_rdd = match_list_rdd.rdd.map { p =>

      val match_time = p.getString(5)
      val match_time_second = sdf.parse(match_time).getTime

      (s"${p.getString(0)}_${p.getString(1)}_${p.getString(2)}_${p.getString(3)}_${p.getString(4)}",
        (p.getString(6), p.getString(7), p.getString(8), match_time_second.toString()))
    }
    //tuple_same_init_odds_rdd.collect().foreach(println)


    // out: ((280_737_2004_瑞典杯_-25,(130695,1099751400000)),30)
    val tuple_same_init_odds_index_rdd = tuple_same_init_odds_rdd.zipWithIndex
    //tuple_same_init_odds_index_rdd.collect().foreach(println)


    // out: (301925_234_1070_8500_17000,(590,(324027,1310058000000)))
    val new_tuple_same_init_odds_index_rdd = tuple_same_init_odds_index_rdd.map(p =>
      //(      p._1._1, ((p._1._2._1, p._2),p._1._2._2)     )
      (p._1._1, (p._2, p._1._2))
    )
    //new_tuple_same_init_odds_index_rdd.collect().foreach(println)


    //OUT: (280_2403_2011/2012_塞浦甲_-25,(147386,(337416,1,0,1321894800000)))
    //倒叙
    val new_tuple_same_init_odds_index_order_rdd = new_tuple_same_init_odds_index_rdd.groupByKey().map { p =>
      val sortArray = p._2.toArray.sortWith(_._1 > _._1)
      (p._1, sortArray)
    }
    //    new_tuple_same_init_odds_index_order_rdd.collect().foreach{p =>
    //          println(p._1)
    //          p._2.foreach(println)
    //        }
    //new_tuple_same_init_odds_index_order_rdd.collect().foreach(println)


    //    //OUT: (280_3108_2014/2015_西班牙杯_-125,462066_462069_459990,3_3_1,0_0_0)
    //    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
    //      (p._1,
    //        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
    //        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
    //        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _))
    //    ).filter(p => p._3.length > mini_cnt)
    //    //new_tuple_same_init_odds_rdd.collect().foreach(println)
    //
    //    /// 保存DB
    //    SaveDds.save_asia_dds_odds_result_to_mysql(new_tuple_same_init_odds_rdd, dds_type)

    //OUT: (280_3108_2014/2015_西班牙杯_-125,462066_462069_459990,3_3_1,0_0_0)
    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
      (p._1,
        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _))
    ).filter(p => p._3.length > mini_cnt).map(p =>
      //(dds_type.toString()+"_"+p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()))
      (p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()))

    new_tuple_same_init_odds_rdd
  }




  // 根据key(companyid_seasonpre_initodds)
  // 统计后面参数(例如: match_id,match_result,asia_init_win_flag)
  def compute_initodds_by_company_seasonname(match_list_rdd:  DataFrame, dds_type: String, mini_cnt: Int): RDD[(String,String)] ={

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    // 转换为元祖 (companyid_seasonpre_initodds, (match_id,match_result,asia_init_win_flag,match_time) )
    // OUT: (280_世外欧洲_25,(130330,1,3,1126108740000))
    val tuple_same_init_odds_rdd = match_list_rdd.rdd.map { p =>

      val match_time = p.getString(3)
      val match_time_second = sdf.parse(match_time).getTime

      (s"${p.getString(0)}_${p.getString(1)}_${p.getString(2)}",
        (p.getString(4), p.getString(5), p.getString(6), match_time_second.toString()))
    }
    //tuple_same_init_odds_rdd.collect().foreach(println)


    // out: ((280_瑞典杯_-25,(130695,1099751400000)),30)
    val tuple_same_init_odds_index_rdd = tuple_same_init_odds_rdd.zipWithIndex
    //tuple_same_init_odds_index_rdd.collect().foreach(println)


    // out:
    val new_tuple_same_init_odds_index_rdd = tuple_same_init_odds_index_rdd.map(p =>
      //(      p._1._1, ((p._1._2._1, p._2),p._1._2._2)     )
      (p._1._1, (p._2, p._1._2))
    )
    //new_tuple_same_init_odds_index_rdd.collect().foreach(println)


    //OUT: (280_塞浦甲_-25,(147386,(337416,1,0,1321894800000)))
    //倒叙
    val new_tuple_same_init_odds_index_order_rdd = new_tuple_same_init_odds_index_rdd.groupByKey().map { p =>
      val sortArray = p._2.toArray.sortWith(_._1 > _._1)
      (p._1, sortArray)
    }
    //    new_tuple_same_init_odds_index_order_rdd.collect().foreach{p =>
    //          println(p._1)
    //          p._2.foreach(println)
    //        }
    //new_tuple_same_init_odds_index_order_rdd.collect().foreach(println)


    //OUT: (280_西班牙杯_-125,462066_462069_459990,3_3_1,0_0_0)
    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
      (p._1,
        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _))
    ).filter(p => p._3.length > mini_cnt).map(p =>
//      (dds_type.toString()+"_"+p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString())
//      )
      (p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString())
      )
    //new_tuple_same_init_odds_rdd.collect().foreach(println)

    //SaveDds.save_asia_dds_odds_result_to_mysql(new_tuple_same_init_odds_rdd, dds_type)

    new_tuple_same_init_odds_rdd
  }







  // 根据key(companyid_seasonid_seasonnamepre_seasonpre_initodds_currodds)
  // 统计后面参数(例如: match_id,match_result,asia_init_win_flag,asia_curr_win_flag)
  def compute_odds_by_company_seasonid(match_list_rdd:  DataFrame, dds_type: String, mini_cnt: Int): RDD[(String,String)] ={

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    //    selectExpr("company_id", "season_id", "season_name_pre", "season_pre", "init_odds", "curr_odds", "match_time", "match_id","match_result", "assia_init_win_flag", "asia_curr_win_flag")
    //

    // 转换为元祖 (companyid_seasonid_seasonnamepre_seasonpre_initodds, (match_id,match_result,asia_init_win_flag,match_time) )
    // OUT:
    val tuple_same_init_odds_rdd = match_list_rdd.rdd.map { p =>

      val match_time = p.getString(6)
      val match_time_second = sdf.parse(match_time).getTime

      (s"${p.getString(0)}_${p.getString(1)}_${p.getString(2)}_${p.getString(3)}_${p.getString(4)}_${p.getString(5)}",
        (p.getString(7), p.getString(8), p.getString(9), p.getString(10), match_time_second.toString()))
    }
    //tuple_same_init_odds_rdd.collect().foreach(println)


    // out:
    val tuple_same_init_odds_index_rdd = tuple_same_init_odds_rdd.zipWithIndex
    //tuple_same_init_odds_index_rdd.collect().foreach(println)


    // out:
    val new_tuple_same_init_odds_index_rdd = tuple_same_init_odds_index_rdd.map(p =>
      //(      p._1._1, ((p._1._2._1, p._2),p._1._2._2)     )
      (p._1._1, (p._2, p._1._2))
    )
    //new_tuple_same_init_odds_index_rdd.collect().foreach(println)


    //OUT:
    //倒叙
    val new_tuple_same_init_odds_index_order_rdd = new_tuple_same_init_odds_index_rdd.groupByKey().map { p =>
      val sortArray = p._2.toArray.sortWith(_._1 > _._1)
      (p._1, sortArray)
    }
    //    new_tuple_same_init_odds_index_order_rdd.collect().foreach{p =>
    //          println(p._1)
    //          p._2.foreach(println)
    //        }
    //new_tuple_same_init_odds_index_order_rdd.collect().foreach(println)


    //OUT:
    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
      (p._1,
        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._4.toString()).reduce(_ + "" + _))
    ).filter(p => p._3.length > mini_cnt).map(p =>
//      (dds_type.toString()+"_"+p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString())
//      )
      (p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString())
      )
    //new_tuple_same_init_odds_rdd.collect().foreach(println)

    //SaveDds.save_asia_init_curr_dds_odds_result_to_mysql(new_tuple_same_init_odds_rdd, dds_type)

    new_tuple_same_init_odds_rdd
  }





  // 根据key(companyid_seasonid_seasonnamepre_seasonpre_initodds_currodds)
  // 统计后面参数(例如: match_id,match_result,asia_init_win_flag,asia_curr_win_flag)
  def compute_odds_by_seasonname(match_list_rdd:  DataFrame, dds_type: String, mini_cnt: Int): RDD[(String,String)] ={

    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    // 转换为元祖 (companyid_seasonpre_initodds_currodds, (match_id,match_result,asia_init_win_flag,asia_curr_win_flag,match_time) )
    // OUT:
    val tuple_same_init_odds_rdd = match_list_rdd.rdd.map { p =>

      val match_time = p.getString(4)
      val match_time_second = sdf.parse(match_time).getTime

      (s"${p.getString(0)}_${p.getString(1)}_${p.getString(2)}_${p.getString(3)}",
        (p.getString(5), p.getString(6), p.getString(7), p.getString(8), match_time_second.toString()))
    }
    //tuple_same_init_odds_rdd.collect().foreach(println)


    // out:
    val tuple_same_init_odds_index_rdd = tuple_same_init_odds_rdd.zipWithIndex
    //tuple_same_init_odds_index_rdd.collect().foreach(println)


    // out:
    val new_tuple_same_init_odds_index_rdd = tuple_same_init_odds_index_rdd.map(p =>
      //(      p._1._1, ((p._1._2._1, p._2),p._1._2._2)     )
      (p._1._1, (p._2, p._1._2))
    )
    //new_tuple_same_init_odds_index_rdd.collect().foreach(println)


    //OUT:
    //倒叙
    val new_tuple_same_init_odds_index_order_rdd = new_tuple_same_init_odds_index_rdd.groupByKey().map { p =>
      val sortArray = p._2.toArray.sortWith(_._1 > _._1)
      (p._1, sortArray)
    }
    //    new_tuple_same_init_odds_index_order_rdd.collect().foreach{p =>
    //          println(p._1)
    //          p._2.foreach(println)
    //        }
    //new_tuple_same_init_odds_index_order_rdd.collect().foreach(println)


    //OUT:
    val new_tuple_same_init_odds_rdd = new_tuple_same_init_odds_index_order_rdd.map(p =>
      (p._1,
        p._2.map(p => p._2._1.toString()).reduce(_ + "," + _),
        p._2.map(p => p._2._2.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._3.toString()).reduce(_ + "" + _),
        p._2.map(p => p._2._4.toString()).reduce(_ + "" + _))
    ).filter(p => p._3.length > mini_cnt).map(p =>
//      (dds_type.toString()+"_"+p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString())
//      )
      (p._1.toString(), p._2.toString()+";"+p._3.toString()+";"+p._4.toString()+";"+p._5.toString())
      )
    //new_tuple_same_init_odds_rdd.collect().foreach(println)

    //SaveDds.save_asia_init_curr_dds_odds_result_to_mysql(new_tuple_same_init_odds_rdd, dds_type)

    new_tuple_same_init_odds_rdd
  }





}


